package com.bjtu.boatmq.broker.service.request;

import cn.boat.router.consumer.ConsumerMetaData;
import cn.boat.router.consumer.PartitionData;
import cn.boat.router.consumer.TopicWithPartitionData;
import cn.boat.router.models.TopicAndPartition;
import com.bjtu.boatmq.broker.config.BrokerConfig;
import com.bjtu.boatmq.broker.service.outer.Broker2RouterService;
import com.bjtu.boatmq.broker.service.outer.Broker2StoreService;
import com.bjtu.boatmq.broker.service.outer.BrokerLogger;
import com.bjtu.boatmq.common.protocol.Message;
import com.bjtu.boatmq.common.protocol.PacketSource;
import com.bjtu.boatmq.common.protocol.ResponseCode;
import com.bjtu.boatmq.common.protocol.request.PullMessageRequest;
import io.netty.channel.ChannelHandlerContext;

import java.util.List;

public class RequestService {
    private int brokerId;   // 当前broker的id
    // 当前broker的元数据
    private static List<TopicAndPartition> topicAndPartitionList;
    // 消费者订阅主题的元数据
    private static List<ConsumerMetaData> consumerMetaDataList;

    public RequestService() {
        brokerId = BrokerConfig.getBrokerId();
    }

    public PacketSource createResponsePacket(int requestId) {
        PacketSource responsePacket = new PacketSource();
        responsePacket.setRequestId(requestId);             // 响应对应的请求
        responsePacket.setFlag(PacketSource.RESPONSE);      // 数据包类型为响应
        responsePacket.setSponsorType(PacketSource.BROKER); // 返回类型为broker
        responsePacket.setSponsorId(brokerId + "");           // 响应发起者为当前broker
        return responsePacket;
    }

    /**
     * 发送消息请求
     */
    public boolean sendMessageRequest(ChannelHandlerContext ctx, PacketSource ps) {
        System.out.println("处理请求:" + ps.getRequestId());
        // 创建响应数据包
        PacketSource responsePacket = createResponsePacket(ps.getRequestId());
        try {
            // 0.更新元数据
            topicAndPartitionList = Broker2RouterService.getBrokerMetaData();

            // 1.解析Message对象，可能是message列表
            Message message = (Message) ps.getData();
            System.out.println(message.toString());
            // 2.执行消息过滤，判断是否是当前分区leader
            if (!isPartitionLeader(message.getTopic(), message.getPartitionId())) {
                // 不满足，直接返回响应，写上报错信息
                responsePacket.setCode(ResponseCode.ERR_NO_PARTITION_LEADER);
                responsePacket.setRemark("当前broker不是消息的分区leader");
                dealWithResponse(ctx, responsePacket);
                return false;
            }
            // 3.存储消息，返回消息偏移量
            int offset = Broker2StoreService.storeMessage(message);
            if (offset > 0) { // 存储成功
                responsePacket.setCode(ResponseCode.NO_ERROR);
                responsePacket.setRemark("offset:"+offset);
                dealWithResponse(ctx,responsePacket);
                return true;
            } else { // 存储失败
                responsePacket.setCode(ResponseCode.ERR_STORE_MESSAGE);
                responsePacket.setRemark("持久化消息失败");
                dealWithResponse(ctx,responsePacket);
                return false;
            }
        } catch (Exception e) {
            responsePacket.setCode(ResponseCode.ERR_STORE_MESSAGE);
            responsePacket.setRemark("持久化消息时发生意外错误");
            dealWithResponse(ctx,responsePacket);
            return false;
        }
    }

    /**
     * 拉取消息请求
     */
    public boolean pullMessageRequest(ChannelHandlerContext ctx, PacketSource ps) {
        // 创建响应数据包
        PacketSource responsePacket = createResponsePacket(ps.getRequestId());
        try {
            // 0.定义请求数据
            String consumerId = ps.getSponsorId();
            String topic = "";
            int partitionId = 0;
            int requestOffset = 0;
            // 1.获取broker元数据/consuemer元数据
            topicAndPartitionList = Broker2RouterService.getBrokerMetaData();
            consumerMetaDataList = Broker2RouterService.getConsuemrMetaData();
            // 2. 解析数据包
            PullMessageRequest pullMessageRequest = new PullMessageRequest();
            if(ps.getData() instanceof PullMessageRequest){
                topic = pullMessageRequest.getTopicName();
                partitionId = pullMessageRequest.getPartitionId();
                requestOffset = pullMessageRequest.getConsumeOffset();
            }else{
                // TODO：报错
            }
            // 3.判断是否当前分区leader
            if (!isPartitionLeader(topic, partitionId)) {
                responsePacket.setCode(ResponseCode.ERR_NO_PARTITION_LEADER);
                responsePacket.setRemark("当前Broker不是分区leader");
                dealWithResponse(ctx,responsePacket);
                return false;
            }
            // 4.判断消费者是否订阅topic-partition
            if (!isSubscribePartition(consumerId, topic, partitionId)) {
                responsePacket.setCode(ResponseCode.ERR_NO_SUBSCRIBE_TOPIC);
                responsePacket.setRemark("未订阅此主题/分区");
                dealWithResponse(ctx,responsePacket);
                return false;
            }
            // 加载拉取的偏移量
            if (requestOffset == -1) {//从最新开始获取
                // 获取分区的存储偏移量
                int partitionOffset = Broker2StoreService.getPartitionOffset(topic, partitionId);
                // 记录到消费偏移量中
                Broker2StoreService.storeConsumeOffset(consumerId, topic, partitionId, partitionOffset);
            }else if(requestOffset == -2){// 从存储的偏移量开始
                requestOffset = Broker2StoreService.getConsumeOffset(consumerId,topic,partitionId);
            }
            // 5.从指定偏移量开始获取数据，放入List<Message>
            List<Message> backData = Broker2StoreService.getMessageList(topic, partitionId, requestOffset);
            // 6.封装成数据包，返回给消费者
            if (backData != null) {
                responsePacket.setCode(ResponseCode.NO_ERROR);
                responsePacket.setData(backData);
                responsePacket.setRemark("成功获取消息");
                dealWithResponse(ctx,responsePacket);
                return true;
            } else {
                responsePacket.setCode(ResponseCode.ERR_NO_FOUND_MESSAGE);
                responsePacket.setRemark("未找到对应消息");
                dealWithResponse(ctx,responsePacket);
                return false;
            }
        } catch (Exception e) {
            BrokerLogger.write(e);
            responsePacket.setCode(ResponseCode.ERR_PULL_EXCEPTION);
            responsePacket.setRemark("拉取消息发生意外错误");
            dealWithResponse(ctx,responsePacket);
            return false;
        }
    }

    /**
     * 提交偏移量请求
     */
    public boolean commitOffsetRequest(ChannelHandlerContext ctx, PacketSource ps) {
        // 创建响应数据包
        PacketSource responsePacket = createResponsePacket(ps.getRequestId());
        try {
            // 1. 获取consumerId, topic, PartitionId, HaveConsumeOffset
            String consumerId = ps.getSponsorId();
            String topic = "";
            int partitionId = 0;
            int haveConsumeOffset = 0;
            // 2. 判断是否满足分区leader
            if(!isPartitionLeader(topic,partitionId)){
                responsePacket.setCode(ResponseCode.ERR_NO_PARTITION_LEADER);
                responsePacket.setRemark("当前Broker不是分区leader");
                dealWithResponse(ctx,responsePacket);
                return false;
            }
            // 3. 存储偏移量
            int commitRes = Broker2StoreService.storeConsumeOffset(consumerId,topic,partitionId,haveConsumeOffset);
            if(commitRes > 0){
                responsePacket.setCode(ResponseCode.NO_ERROR);
                responsePacket.setRemark("提交偏移量成功");
                dealWithResponse(ctx,responsePacket);
                return true;
            }else{
                responsePacket.setCode(ResponseCode.ERR_STORE_CONSUME_OFFSET);
                responsePacket.setRemark("存储消费偏移量失败");
                dealWithResponse(ctx,responsePacket);
                return false;
            }
        } catch (Exception e) {
            e.printStackTrace();
            BrokerLogger.write(e);
            responsePacket.setCode(ResponseCode.ERR_COMMIT_EXCEPTION);
            responsePacket.setRemark("提交偏移量时发生意外错误");
            dealWithResponse(ctx,responsePacket);
            return false;
        }
    }

    /**
     * 未知请求
     */
    public boolean unknowRequest(ChannelHandlerContext ctx, PacketSource ps) {
        // 创建响应数据包
        PacketSource responsePacket = createResponsePacket(ps.getRequestId());
        responsePacket.setCode(ResponseCode.ERR_UNKNOWN_REQUEST);
        responsePacket.setRemark("unknown request code.");
        dealWithResponse(ctx, responsePacket);
        return true;
    }

    /**
     * 返回响应
     *
     * @param ctx
     * @param responsePacket
     */
    public void dealWithResponse(ChannelHandlerContext ctx, PacketSource responsePacket) {
        System.out.println("返回响应：" + responsePacket.getRequestId());
        ctx.writeAndFlush(responsePacket);
    }

    /**
     * 当前broker是否是message的分区leader
     */
    public boolean isPartitionLeader(String topic, int partitionId) {
        /*for(TopicAndPartition item : topicAndPartitionList){
            item.toString();
            // TODO:根据message的topic和partition，找寻leader是不是当前的brokerId
        }*/
        return true;
    }

    /**
     * 消费者是否订阅topic-partition
     *
     * @param consumerId
     * @param topic
     * @param partitionId
     * @return
     */
    public boolean isSubscribePartition(String consumerId, String topic, int partitionId){
        // 判断当前请求的消费者，是否订阅topic-partition
        // TODO:等待京诚完成
        for (ConsumerMetaData item : consumerMetaDataList) {
            if (item.getConsumerId().equals(consumerId)) {
                // 开始查看是否订阅topic
                for (TopicWithPartitionData topicItem : item.getTopicWithPartitionDataList()) {
                    if (topicItem.getTopic().equals(topic)) {
                        // 查看是否消费该分区
                        for (PartitionData partitionItem : topicItem.getPartitionDataList()) {
                            if (partitionItem.getPartitionLeader() == brokerId) {
                                return true;
                            }
                        }
                        break;  // 找到topic 退出topic遍历
                    }
                }
                break;  // 找到consumer 退出consumer遍历
            }
        }
        return false;
    }
}
